package com.haiyou.data.common.accessor.redis.redisson.mapping;

import java.io.Serializable;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiFunction;
import java.util.function.Function;

import org.redisson.api.ExpiredObjectListener;
import org.redisson.api.LocalCachedMapOptions;
import org.redisson.api.MapCacheOptions;
import org.redisson.api.MapOptions;
import org.redisson.api.RLocalCachedMap;
import org.redisson.api.RMap;
import org.redisson.api.RMapCache;
import org.redisson.api.map.MapWriter;

import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.haiyou.common.entity.Entity;
import com.haiyou.common.util.reflect.AnnotationFinder;
import com.haiyou.data.common.accessor.annotation.MappingType;
import com.haiyou.data.common.accessor.annotation.NodePolicy;
import com.haiyou.data.common.accessor.redis.redisson.RedissonAccessor;
import com.haiyou.data.common.accessor.rsync.Synchronizer;
import com.haiyou.data.common.cache.redis.RKeys;
import com.haiyou.data.common.cache.redis.redisson.RedissonAtomicExecutor;
import com.haiyou.data.common.persister.mysql.persister.Persister;

import lombok.extern.slf4j.Slf4j;

/**
 * 使用redisson映射数据
 * </p>
 * 注意:MapLoader 无法设置过期时间 暂时不使用<br>
 * RMapCache 某些情况下 使用MapLoader 加载缓存中没有的数据不会执行load,<br>
 * 存在过多数据量是使用过期key设计 由于redis hash中无法对某个key设置超时 redisson
 * 使用的客户端实现,在redis中创建定时redisson__timeout__set: 对象
 * 
 * @author xingyuan
 * @author xingyuan
 * @param <V>
 */
@Slf4j
public abstract class MappingAccessor<K extends Serializable & Comparable<K>, V extends Entity<K>>
		extends RedissonAccessor<K, V> {

	private static final String SUFFIX = "MappingView";

	protected RMap<K, V> mapping;

	protected Persister<K, V> persister;
	
	protected Synchronizer<K, V> synchronizer;

	protected int nodeSize;// mapping节点个数

//	private final MapLoader<K, V> mapLoader = new MapLoader<K, V>() {
//		@Override
//		public V load(K key) {
//			V value = null;
//			try {
//				return value = persister.query(key);
//			} finally {
//				if (value != null) {
//					synchronizer.flush(key, value);
//				}
//			}
//		}
//
//		@Override
//		public Iterable<K> loadAllKeys() {
//			return persister.queryIds();
//		}
//	};

	protected final MapWriter<K, V> mapWriter = new MapWriter<K, V>() {
		@Override
		public void write(Map<K, V> map) {
			synchronizer.flush(map);
		}

		@Override
		public void delete(Collection<K> keys) {
			
		}
	};

	protected RMap<K, V> createMap() {
		MapOptions<K, V> mapOptions = MapOptions.defaults();
		// mapOptions.loader(mapLoader);
		mapOptions.writer(mapWriter);
		return redissonClient.getMap(RKeys.createRKey(RMap.class, keyPrefix, SUFFIX), mapOptions);
	}

	protected RMapCache<K, V> createMapCache() {
		MapCacheOptions<K, V> mapOptions = MapCacheOptions.defaults();
		// mapOptions.loader(mapLoader);
		mapOptions.writer(mapWriter);
		return new RedissonMapCacheWrapper<>(redissonClient, keyClass,
				RKeys.createRKey(RMapCache.class, keyPrefix, SUFFIX), mapOptions, nodeSize);
	}

	protected RLocalCachedMap<K, V> createLocalCachedMap() {
		LocalCachedMapOptions<K, V> localCachedMapOptions = LocalCachedMapOptions.defaults();
		// localCachedMapOptions.loader(mapLoader);
		localCachedMapOptions.writer(mapWriter);
		return redissonClient.getLocalCachedMap(RKeys.createRKey(RLocalCachedMap.class, keyPrefix, SUFFIX),
				localCachedMapOptions);
	}

	protected RMap<K, V> createMapping() {
		MappingType mappingType = AnnotationFinder.findAnnotation(this.getClass(), MappingType.class);
		String mappingClassType = mappingType == null ? MappingType.MAPPING_CACHE : mappingType.value();

		RMap<K, V> mapping = mappingClassType.equals(MappingType.MAPPING) ? createMap()
				: mappingClassType.equals(MappingType.MAPPING_CACHE) ? createMapCache()
						: mappingClassType.equals(MappingType.LOCAL_CACHED_MAPPING) ? createLocalCachedMap() : null;
		mapping.addListener(new ExpiredObjectListener() {
			@Override
			public void onExpired(String name) {

			}
		});

		return mapping;
	}

	@Override
	public void init() {
		super.init();
		persister = Persister.getInstance(valueClass);
		synchronizer = new Synchronizer<>(this, persister);
		synchronizer.init();
		timeToLive = synchronizer.getTimeToLive();
		timeToLiveUnit = synchronizer.getTimeToLiveUnit();
		NodePolicy nodeNolicy = AnnotationFinder.findAnnotation(this.getClass(), NodePolicy.class);
		this.nodeSize = nodeNolicy == null ? RedissonMapCacheWrapper.DEFAULT_NODE_SIZE : nodeNolicy.nodeSize();
		mapping = createMapping();
	}

	@Override
	public void destory() {
		super.destory();
		synchronizer.destory();
	}

	/**
	 * 注意: 此处不调用 synchronizer.flush(K,V)
	 */
	@Override
	public Map<K, V> getAll(Iterable<? extends K> keys) {
		// log.info("getAll keys:{}",keys);
		// redis中数据

		Set<K> keySet = Sets.newHashSet(keys);

		Map<K, V> maps = mapping.getAll(Sets.newHashSet(keys));

		// 移除redis中 已有
		keySet.removeAll(maps.keySet());

		if (!keySet.isEmpty()) {
			// 查询数据库
			maps.putAll(queryAll(keySet));
		}

		return maps;
	}

	@SuppressWarnings("unchecked")
	@Override
	public boolean containsKey(Object key) {
		return mapping.containsKey(key) || persister.contains((K) key);
	}

	@SuppressWarnings("unchecked")
	@Override
	public boolean containsValue(Object value) {
		return get(((V) value).getId()).equals(value);
	}

	@SuppressWarnings("unchecked")
	@Override
	public V get(Object key) {
		V v = null;
		try {
			v = mapping.get(key);

			if (v == null) {
				v = query((K) key);
				return v;
			}
			return v;
		} finally {
//			if (v != null) {
//				synchronizer.flush((K) key, v);
//			}
		}
	}

	/**
	 * 通过数据库查询
	 * 
	 * @param key
	 * @return
	 */
	public V query(K key) {

		V v = persister.query(key);

		if (v != null) {
			put(key, v);
		}
		return v;
	}

	/**
	 * 批量查询
	 * 
	 * @param keys
	 * @return
	 */
	public Map<K, V> queryAll(Set<K> keys) {

		List<V> vs = persister.queryBatch(keys);

		Map<K, V> map = Maps.newHashMap();

		for (V v : vs) {
			Entity<K> entity = (Entity<K>) v;
			map.put(entity.getId(), v);
			put(entity.getId(), v);
		}

		return map;
	}

	@Override
	public V put(K key, V value) {
		if (timeToLive > 0 && mapping instanceof RMapCache) {
			RMapCache<K, V> mapCache = (RMapCache<K, V>) mapping;
			return mapCache.put(key, value, timeToLive, timeToLiveUnit);
		} else {
			return mapping.put(key, value);
		}
	}

	@Override
	public V remove(Object key) {
		return mapping.remove(key);
	}

	@Override
	public void removeAll(Iterable<? extends K> keys) {
		mapping.fastRemove(Iterables.toArray(keys, keyClass));
		synchronizer.giveUp(keys);
		persister.delete(keys);
		if (keyRecycler != null)
			keyRecycler.retrieve(Iterables.toArray(keys, keyClass));
	}

	@Override
	public void putAll(Map<? extends K, ? extends V> m) {
		if (timeToLive > 0 && mapping instanceof RMapCache) {
			RMapCache<K, V> mapCache = (RMapCache<K, V>) mapping;
			mapCache.putAll(m, timeToLive, timeToLiveUnit);
		} else {
			// RMapCache 中 putAll(m)方法为final不能重写
			// 修改成调用可以重写的putAll(m,size),重写后继续调用putAll(m)接口
			mapping.putAll(m, m.size());
		}
	}

	@Override
	public V putIfAbsent(K key, V value) {
		return RedissonAtomicExecutor.call(mapping, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> super.putIfAbsent(key, value));
	}

	@SuppressWarnings("unchecked")
	@Override
	public boolean remove(Object key, Object value) {
		AtomicBoolean bool = new AtomicBoolean();
		RedissonAtomicExecutor.run(mapping, (K) key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> bool.compareAndSet(false, super.remove(key, value)));
		return bool.get();
	}

	@Override
	public boolean replace(K key, V oldValue, V newValue) {
		AtomicBoolean bool = new AtomicBoolean();
		RedissonAtomicExecutor.run(mapping, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> bool.compareAndSet(false, super.replace(key, oldValue, newValue)));
		return bool.get();
	}

	@Override
	public V replace(K key, V value) {
		return RedissonAtomicExecutor.call(mapping, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> super.replace(key, value));
	}

	@Override
	public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) {
		return RedissonAtomicExecutor.call(mapping, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> super.computeIfAbsent(key, mappingFunction));
	}

	@Override
	public V computeIfPresent(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
		return RedissonAtomicExecutor.call(mapping, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> super.computeIfPresent(key, remappingFunction));
	}

	@Override
	public V compute(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
		return RedissonAtomicExecutor.call(mapping, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> super.compute(key, remappingFunction));
	}

	@Override
	public V merge(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction) {
		return RedissonAtomicExecutor.call(mapping, key, lockWaitTime(), lockLeaseTime(), lockTimeUnit(),
				() -> super.merge(key, value, remappingFunction));
	}

	@Override
	public Class<?> getInitClass() {
		return this.getClass();
	}


	
	
}
